/*
 * Copyright 2011 Rolf Aden
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License. 
 */

package com.ec;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.tools.mapreduce.DatastoreMutationPool;
import com.google.appengine.tools.mapreduce.contrib.DatastoreMapper;

/**
 * DatastoreMapper for the renaming of properties
 * 
 * @author Rolf Aden
 */
public class RenamePropertyMapper extends DatastoreMapper {
	
	enum Phase {
		run,
		verify
	}
	
	private static final Logger LOG = Logger.getLogger(RenamePropertyMapper.class.getName());
	
	static final String PHASE = "ec.phase";
	static final String OLD_PROPERTY_NAME = "ec.oldPropertyName";
	static final String NEW_PROPERTY_NAME = "ec.newPropertyName";
	static final String USE_TRANSACTIONS = "ec.useTransactions";
	
	private DatastoreService ds = DatastoreServiceFactory.getDatastoreService();
	private DatastoreMutationPool mutationPool;
	private String oldPropertyName;
	private String newPropertyName;
	private Phase phase;
	private boolean useTransactions = true;
	private ErrorReporter errorReporter;
	
	public void taskSetup(Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException ,InterruptedException {
		super.taskSetup(context);
		
		//	get the configuration properties here, 
		//	because they stay the same as long as the task runs
		this.oldPropertyName = context.getConfiguration().get(OLD_PROPERTY_NAME);
		this.newPropertyName = context.getConfiguration().get(NEW_PROPERTY_NAME);
		this.phase = "verify".equals(context.getConfiguration().get(PHASE))
    		? Phase.verify
    		: Phase.run;

		this.useTransactions = Boolean.parseBoolean(context.getConfiguration().get(USE_TRANSACTIONS));
		
		//	check with appengine-mapreduce-java team how to communicate a fatal error, 
		//	meaning it would make no sense to proceed with the task, e.g. no property name given
		//	refer to comment in MapReduceServlet#handleMapperWorker, currently it's not possible
		validate();
		
		this.mutationPool = this.getAppEngineContext(context).getMutationPool();
		this.errorReporter = new ErrorReporter(context, ds);
	}

	@Override
	public void map(Key key, Entity entity, Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException, InterruptedException 
	{
		if (LOG.isLoggable(Level.FINER)) {
			LOG.finer("Processing key: " + key + " in phase " + this.phase);
		}
		
		switch (this.phase) {
			case run:
	    		if (!entity.hasProperty(this.newPropertyName))
	    		{
	    			if (entity.hasProperty(this.oldPropertyName)) {
	    				if (entity.isUnindexedProperty(this.oldPropertyName)) {
	    					entity.setUnindexedProperty(this.newPropertyName, entity.getProperty(this.oldPropertyName));
	    				} else {
	    					entity.setProperty(this.newPropertyName, entity.getProperty(this.oldPropertyName));
	    				}
	        			entity.removeProperty(this.oldPropertyName);
	    			} else {
	    				entity.setProperty(this.newPropertyName, null);
	    			}
	    			put(entity);	
	    		} else {
	    			if (entity.hasProperty(this.oldPropertyName)) {
	        			entity.removeProperty(this.oldPropertyName);
	        			put(entity);	
	    			}
	    		}
				break;
				
			case verify:
				if (!entity.hasProperty(this.newPropertyName)) {
					this.errorReporter.addErrorMessage("Entity " + key + ": property " + this.newPropertyName + " is missing");
				}
				if (entity.hasProperty(this.oldPropertyName)) {
					this.errorReporter.addErrorMessage("Entity " + key + ": has still property " + this.oldPropertyName);
				}
				break;
		}
	}
	
	private void put(Entity entity) {
		if (this.useTransactions) {
			Transaction tx = this.ds.beginTransaction();
			try {
				this.ds.put(entity);
				tx.commit();
			} finally {
				if (tx.isActive()) {
					tx.rollback();
				}
			}
		} else {
			this.mutationPool.put(entity);	
		}
	}
	
	protected void validate() throws IOException {
		StringBuilder bld = null;
		if (this.oldPropertyName == null || this.oldPropertyName.trim().isEmpty()) {
			bld = new StringBuilder("old Property is missing");
		}
		if (this.newPropertyName == null || this.newPropertyName.trim().isEmpty()) {
			bld = bld == null
				? new StringBuilder("new Property is missing")
				: bld.append("\nnew Property is missing");
		}
		
		if (bld != null) {
			throw new IOException(bld.toString());
		}
	}

}
